package spark.core.rdd

import org.apache.spark.{SparkConf, SparkContext}

object rddActOperators {
  def main(args: Array[String]): Unit = {
    // 环境准备
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rddActOperators")
    val sparkContext: SparkContext = new SparkContext(sparkConf)

    // 业务操作
    // Todo 1. reduce
    // 聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val reduceResult: Int = rdd.reduce(_ + _)

    // Todo 2. collect
    // 在驱动程序中(Driver)，以数组Array的形式返回数据集的所有元素

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    // 收集数据到Driver
    //    rdd.collect().foreach(println)

    // Todo 3. count
    // 返回RDD中元素的个数

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val countResult: Long = rdd.count()
    //    println(countResult)

    // Todo 4. first
    // 返回RDD中的第一个元素

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val firstResult: Int = rdd.first()
    //    println(firstResult)

    // Todo 5. take
    // 返回一个由RDD的前n个元素组成的 数组

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 9, 3, 4))
    //    val takeResult: Array[Int] = rdd.take(2)
    //    takeResult.foreach(println)

    // Todo 6. takeOrdered
    // 返回该RDD排序后的前n个元素组成的数组

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 9, 2, 4))
    //    // 返回RDD中元素的个数
    //    val result: Array[Int] = rdd.takeOrdered(2)
    //    result.foreach(println)

    // Todo 7. aggregate
    // 分区的数据 通过初始值 和分区内的数据进行聚合，然后 再和初始值进行分区间的数据聚合

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4), 8)
    //    //val result: Int = rdd.aggregate(0)(_ + _, _ + _)
    //    val result: Int = rdd.aggregate(10)(_ + _, _ + _)
    //    println(result)

    // Todo 8. fold
    // 折叠操作，aggregate的简化版操作

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val foldResult: Int = rdd.fold(0)(_ + _)
    //    println(foldResult)

    // Todo 9. countByKey
    // 统计每种key的个数

    //    val rdd: RDD[(Int, String)] = sparkContext.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))
    //    val result: collection.Map[Int, Long] = rdd.countByKey()
    //    println(result)

    // Todo 10. save 相关算子
    // 将数据保存到不同格式的文件中

    //    val RDD: RDD[Int] = sparkContext.makeRDD(List(1, 2, 4, 5, 6))
    //    // 保存成Text文件
    //    RDD.saveAsTextFile("output")
    //    // 序列化成对象保存到文件
    //    RDD.saveAsObjectFile("output1")
    //    // 保存成SequenceFile文件
    //    RDD.map((_, 1)).saveAsSequenceFile("output2")

    // Todo 11. foreach
    // 分布式遍历RDD中的每一个元素，调用指定函数

    //    val rdd: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    // 收集后打印 => collect算子会先将数据按照分区收集到Driver端才执行打印操作的
    //    rdd.map(num => num).collect().foreach(println)
    //    println("****************")
    //    // 分布式打印 => foreach算子是在各个Executor上执行 所以数据打印无法预估顺序
    //    rdd.foreach(println)


    // 释放资源
    sparkContext.stop()

  }

}
